package cn.xqapp.GatewayClient;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import org.apache.commons.lang3.time.StopWatch;

public class GatewayClientHandler extends ChannelInboundHandlerAdapter {

    protected ChannelHandlerContext ctx;
    protected boolean active;

    protected boolean echoMessage;
    protected int echoInterval;

    protected long sendStartTime;
    protected int sendMessageCount;
    protected StopWatch sendStopwatch;

    public GatewayClientHandler(){
        super();

        ctx = null;
        active = false;

        echoMessage = false;
        echoInterval = 0;

        sendMessageCount = 0;
        sendStopwatch = new StopWatch();
    }

    public int getSendMessageCount() {
        return sendMessageCount;
    }

    public long getSendMessageNanoTime(){
        return sendStopwatch.getNanoTime();
    }

    public long getSendSpanMilliseconds(){
        return System.currentTimeMillis() - this.sendStartTime;
    }

    public boolean isActive() {
        return active;
    }

    public void start(byte[] message, boolean echoMessage, int echoInterval) {

        if (this.active && ctx != null) {
            this.sendStartTime = System.currentTimeMillis();

            this.echoMessage = echoMessage;
            this.echoInterval = echoInterval;

            if(sendStopwatch.isSuspended()){
                sendStopwatch.resume();
            }else{
                sendStopwatch.start();
            }
            sendMessageCount++;

            ByteBuf msgByteBuf = ctx.alloc().buffer(message.length);
            msgByteBuf.writeBytes(message);

            ctx.writeAndFlush(msgByteBuf);
        } else {
            //ctx not initialized yet. you were too fast. do something here
        }
    }

    public void close(){
        // stop stopwatch
        sendStopwatch.stop();

        if(ctx != null){
            ctx.close();
            this.active = false;
        }
    }

    public void reset(){
        if(sendStopwatch.isStarted()){
            sendStopwatch.stop();
        }

        sendStopwatch.reset();
        sendStopwatch.start();

        sendStartTime = System.currentTimeMillis();
        sendMessageCount = 0;
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.ctx = ctx;
        this.active = true;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        if(sendStopwatch.isStarted()) {
            sendStopwatch.suspend();
        }

//        ByteBuf msgBuf = (ByteBuf)msg;
//        int msgBufSize = msgBuf.readableBytes();
//
//        System.out.println(String.format("client received message, size %d", msgBufSize));

        if(this.echoMessage){
            if(this.echoInterval > 0){
                Thread.sleep(this.echoInterval);
            }

            if(sendStopwatch.isSuspended()) {
                sendStopwatch.resume();
                sendMessageCount++;
            }

            //System.out.println(String.format("echo message, size %d", msgBufSize));
            ctx.writeAndFlush(msg);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        this.active = false;
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // cause.printStackTrace();
        this.close();
    }
}